package com.pengheng.watermark;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class SocketStreamWordCountWithWaterMark {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // IDEA运行时，提供webui，一般用于本地测试
        //需要引入依赖 flink-runtime-web
        // idea运行，不指定并行度，默认就是电脑的线程数
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        DataStreamSource<String> lineDs = env.socketTextStream("192.168.10.100", 7777);
        SingleOutputStreamOperator<Tuple4<String, Long, String, Long>> sum = lineDs
                .flatMap((FlatMapFunction<String, Tuple4<String, Long, String, Long>>) (line, out) -> {
                    String[] words = line.split(" ");

                    for (String word : words) {
                        out.collect(Tuple4.of(word, 1L, "hello", System.currentTimeMillis()));
                    }
                })
                .assignTimestampsAndWatermarks(
                        //添加水印策略，
                        //总共有两种策略 1.forMonotonousTimestamps 2.forBoundedOutOfOrderness
                        // forBoundedOutOfOrderness允许乱序需设置时长
                        // forMonotonousTimestamps不允许乱序，底层也是调用forBoundedOutOfOrderness只是设置的时间为0
                        WatermarkStrategy.<Tuple4<String, Long, String, Long>>forBoundedOutOfOrderness(Duration.ofMinutes(1))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple4<String, Long, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple4<String, Long, String, Long> element, long recordTimestamp) {
                                        return element.f3;
                                    }
                                }))

                .returns(Types.TUPLE(Types.STRING, Types.LONG, Types.STRING))
                .keyBy(data -> data.f0 + "-" + data.f2)
                .sum(1);

        sum.print();

        env.execute();

    }
}
